package com.niodata.dp.task;

import com.niodata.dp.task.az.FlowExecuteResponse;
import com.niodata.dp.task.dao.DynamicExecutionMapper;
import com.niodata.dp.task.entity.DynamicExecution;
import com.niodata.dp.task.entity.JobParams;
import com.niodata.dp.task.entity.PluginExecTemplate;
import com.niodata.dp.task.service.PluginExecTemplateManager;
import com.niodata.dp.task.service.impl.AzBasedTaskManagerV2;
import com.niodata.dp.util.IdUtil;
import java.util.Date;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

/**
 * execute plugin by azkaban.
 */
@Component("azBasedPluginExecutor")
public class AzBasedPluginExecutor implements PluginExecutor {

  private static Logger logger = Logger.getLogger(AzBasedPluginExecutor.class);

  @Autowired
  @Qualifier("azBasedPluginExecTemplateManager")
  private PluginExecTemplateManager execTemplateManager;
  @Autowired
  private DynamicExecutionMapper executionMapper;

  @Autowired
  @Qualifier("azTaskManager")
  private AzBasedTaskManagerV2 taskManager;

  @Override
  public DynamicExecution execute(String name, String version,
      Map params, long userId, long groupId, String dynamicExecName, String description,
      boolean isUserCreate, String integrationType)
      throws Exception {
    PluginExecTemplate template = execTemplateManager.getTemplate(name, version, userId, groupId);
    if (template == null) {
      List<PluginExecTemplate> templates =
          execTemplateManager.getTemplateByPluginName(name, userId, groupId);
      if (templates != null && templates.size() > 0
          && templates.get(templates.size() - 1).getPluginVersion().compareTo(version) > 0) {
        template = templates.get(templates.size() - 1);
      }
    }

    if (template == null) {
      template = execTemplateManager.createTemplate(name, version, userId, groupId);
      logger.info(
          "create exec template:" + template.getPluginName() + ":" + template.getPluginVersion());
    }
    DynamicExecution execution = new DynamicExecution();
    JobParams jobParams = new JobParams();
    jobParams.setParams(params);
    execution.setName(dynamicExecName);
    execution.setDescription(description);
    execution.setParams(jobParams);
    execution.setTaskName(template.getTaskName());
    execution.setOwnerType("GROUP");
    execution.setOwnerId(groupId);
    execution.setId(IdUtil.uuid());
    execution.setStartTime(new Date());
    execution.setStatus("RUNNING");
    execution.setUserCreate(isUserCreate);
    execution.setType(integrationType);
    executionMapper.save(execution);
    FlowExecuteResponse response =
        taskManager.executeTask(template.getTaskName(), execution.getId());
    logger.info(String.format("execute task %s with execution id=%s,execid=%d",
        template.getTaskName(), execution.getId(), response.getExecid()));
    execution.setExecId(response.getExecid());
    executionMapper.updateExecId(execution.getId(), execution.getExecId());
    return execution;
  }
}
